package com.seaboxdata.kafka;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

/**
 * @author
 * @create 2020-08-28 10:20
 **/
@Component
public class BaseStringKafkaConsumer {

    private Map<String, KafkaConsumer<String, String>> consumers;

    private static final boolean isWin = System.getProperty("os.name").toLowerCase().contains("win");

    private Properties props;

    @Value("${kafka.server}")
    private String kafkaServer;

    @Value("${kafka.jaas.config}")
    private String authConfig;

    @Value("${krb5.config}")
    private String krb5Config;

    @Value("${kafka.win.jaas.config}")
    private String authWinConfig;

    @Value("${krb5.win.config}")
    private String krb5WinConfig;


    @Value("${kafka.kerberos.enable}")
    private String isKerberos;

    public BaseStringKafkaConsumer() {
        props = new Properties();
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "60000");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "10");
        props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 90000);
        consumers = new HashMap<>();
    }

    public KafkaConsumer<String, String> getConsumer(String topic) {

        if ("true".equals(isKerberos)) {
            if (isWin) {
                //如果是windows指定本地的配置文件,本地调试使用
                this.authConfig = authWinConfig;
                this.krb5Config = krb5WinConfig;
            }
            System.setProperty("java.security.auth.login.config", authConfig);
            System.setProperty("java.security.krb5.conf", krb5Config);
            props.put("security.protocol", "SASL_PLAINTEXT");
            props.put("sasl.kerberos.service.name", "kafka");
            props.put("sasl.mechanism", "GSSAPI");
        }

        if (consumers.get(topic) == null) {
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
            props.put(ConsumerConfig.GROUP_ID_CONFIG, topic + "-group");
            KafkaConsumer<String, String> stringStringKafkaConsumer = new KafkaConsumer<>(props);
            stringStringKafkaConsumer.subscribe(Collections.singletonList(topic));
            consumers.put(topic, stringStringKafkaConsumer);
            return stringStringKafkaConsumer;
        } else {
            return consumers.get(topic);
        }
    }
}
